﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using XcoAppSpaces.Core;
using XcoAppSpaces.Core.Ports;
using System.Threading;
using NUnit.Framework;

namespace test.XcoAppSpaces.Core
{
	#region Workers
	class SequentialityWorkerContract : Port<int>
	{
		protected Action<int> intReceived;
		
		public SequentialityWorkerContract()
		{}

		public SequentialityWorkerContract(Action<int> intReceived)
		{
			this.intReceived = intReceived;
		}
	}

	class ExclusiveSequentialWorker : SequentialityWorkerContract
	{
		public ExclusiveSequentialWorker(Action<int> intReceived)
			: base(intReceived)
		{ }

		[XcoExclusive]
		private void Process(int item)
		{
			intReceived(item);
		}
	}

	class ExclusiveParallelWorker : SequentialityWorkerContract
	{
		public ExclusiveParallelWorker(Action<int> intReceived)
			: base(intReceived)
		{ }

		[XcoExclusive(MessageTransferMode =  MessageTransferMode.Parallel)]
		private void Process(int item)
		{
			intReceived(item);
		}
	}

	class ConcurrentParallelWorker : SequentialityWorkerContract
	{
		public ConcurrentParallelWorker(Action<int> intReceived)
			: base(intReceived)
		{ }

		[XcoConcurrent]
		private void Process(int item)
		{
			intReceived(item);
		}
	}

	class ConcurrentSequentialWorker : SequentialityWorkerContract
	{
		public ConcurrentSequentialWorker(Action<int> intReceived)
			: base(intReceived)
		{ }

		[XcoConcurrent(MessageTransferMode = MessageTransferMode.Sequential)]
		private void Process(int item)
		{
			intReceived(item);
		}
	}
	#endregion

	[TestFixture]
	public class TestSequentiality
	{
		private const int postCount = 1000;

		[Test]
		public void TestPortSequentiality()
		{
			using (XcoAppSpace space1 = new XcoAppSpace("tcp.port=8000"),
				space2 = new XcoAppSpace("tcp.port=8001"))
			{
				var port = new Port<int>();
				space1.RunWorker(port);
				int receivedCount = 0;
				List<int> receivedItems = new List<int>();
				port.RegisterReceiver(msg =>
				                     	{
											//Console.WriteLine(msg);
				                     		//Assert.AreEqual(receivedCount, msg);
											receivedItems.Add(msg);
				                     		receivedCount++;
				                     	}, new PortReceiverOptions{MaxDegreeOfParallelism = 1});

				var remotePort = space2.ConnectWorker<Port<int>>(Environment.MachineName + ":8000");

				List<int> postedItems = new List<int>();
				for (int i = 0; i < postCount; i++)
				{
					remotePort.Post(i);
					postedItems.Add(i);
				}
				Thread.Sleep(1000);

                Assert.AreEqual(postCount, receivedCount);
				Assert.AreEqual(postedItems, receivedItems); //data sequentiality must be guaranteed!!
			}
		}

		[Test]
		public void TestExclusiveSequentialWorker()
		{
			using (XcoAppSpace space1 = new XcoAppSpace("tcp.port=8000"),
				space2 = new XcoAppSpace("tcp.port=8001"))
			{
				int receivedCount = 0;
				List<int> receivedItems = new List<int>();
				var worker = new ExclusiveSequentialWorker(msg =>
				                                           	{
				                                           		receivedItems.Add(msg);
				                                           		receivedCount++;
				                                           	});
				space1.RunWorker<SequentialityWorkerContract>(worker);
				var remoteWorker = space2.ConnectWorker<SequentialityWorkerContract>(Environment.MachineName + ":8000");

				List<int> postedItems = new List<int>();
				for (int i = 0; i < postCount; i++)
				{
					remoteWorker.Post(i);
					postedItems.Add(i);
				}
				Thread.Sleep(1000);

				Assert.AreEqual(postCount, receivedCount);
				Assert.AreEqual(postedItems, receivedItems); //data sequentiality must be guaranteed!!
			}
		}

		[Test]
		public void TestExclusiveParallelWorker()
		{
			using (XcoAppSpace space1 = new XcoAppSpace("tcp.port=8000"),
				space2 = new XcoAppSpace("tcp.port=8001"))
			{
				int receivedCount = 0;
				List<int> receivedItems = new List<int>();
				var worker = new ExclusiveParallelWorker(msg =>
				{
					receivedItems.Add(msg);
					receivedCount++;
				});
				space1.RunWorker<SequentialityWorkerContract>(worker);
				var remoteWorker = space2.ConnectWorker<SequentialityWorkerContract>(Environment.MachineName + ":8000");

				List<int> postedItems = new List<int>();
				for (int i = 0; i < postCount; i++)
				{
					remoteWorker.Post(i);
					postedItems.Add(i);
				}
				Thread.Sleep(1000);

				Assert.AreEqual(postCount, receivedCount);
				Assert.AreNotEqual(postedItems, receivedItems); //data sequentiality NOT guaranteed
			}
		}

		[Test]
		public void TestConcurrentParallelWorker()
		{
			using (XcoAppSpace space1 = new XcoAppSpace("tcp.port=8000"),
				space2 = new XcoAppSpace("tcp.port=8001"))
			{
				int receivedCount = 0;
				List<int> receivedItems = new List<int>();
				var worker = new ConcurrentParallelWorker(msg =>
				{
					receivedItems.Add(msg);
					receivedCount++;
				});
				space1.RunWorker<SequentialityWorkerContract>(worker);
				var remoteWorker = space2.ConnectWorker<SequentialityWorkerContract>(Environment.MachineName + ":8000");

				List<int> postedItems = new List<int>();
				for (int i = 0; i < postCount; i++)
				{
					remoteWorker.Post(i);
					postedItems.Add(i);
				}
				Thread.Sleep(1000);

				Assert.AreEqual(postCount, receivedCount);
				Assert.AreNotEqual(postedItems, receivedItems); //data sequentiality NOT guaranteed
			}
		}

		[Test]
		public void TestConcurrentSequentialWorker()
		{
			using (XcoAppSpace space1 = new XcoAppSpace("tcp.port=8000"),
				space2 = new XcoAppSpace("tcp.port=8001"))
			{
				int receivedCount = 0;
				List<int> receivedItems = new List<int>();
				var worker = new ConcurrentSequentialWorker(msg =>
				{
					receivedItems.Add(msg);
					receivedCount++;
				});
				space1.RunWorker<SequentialityWorkerContract>(worker);
				var remoteWorker = space2.ConnectWorker<SequentialityWorkerContract>(Environment.MachineName + ":8000");

				List<int> postedItems = new List<int>();
				for (int i = 0; i < postCount; i++)
				{
					remoteWorker.Post(i);
					postedItems.Add(i);
				}
				Thread.Sleep(1000);

				Assert.AreEqual(postCount, receivedCount);
				//Assert.AreNotEqual(postedItems, receivedItems); //data sequentiality NOT guaranteed (but possible, so don't check for non-equality)
			}
		}
	}
}
